package com.utstarcom.hdfs;

import com.utstarcom.util.Common;
import com.utstarcom.util.HDFSConf;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionCodecFactory;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.IOException;
import java.io.InputStream;
import java.net.URI;

/**
 * Created by Venn on 2018/6/2.
 * <p>
 * CompressFile : 解压hdfs上指定目录
 * default format is gzip
 * <p>
 * 1.4G
 * 141 ms
 */
public class DeCompressFile {
    private static Logger logger = LogManager.getLogger(DeCompressFile.class.getName());

    public static void main(String[] args) {

        if (args.length < 1) {
            logger.error("please input the path");
            System.exit(-1);
        }
        String path = args[0];
//        String path = "/aaa/test";
        String hdfsPath = Common.getCommonMap().get("hdfs.uri") + path;


        System.out.println(hdfsPath);

        try {
            long startTime = System.currentTimeMillis();

            getFile(hdfsPath);

            System.out.println("cost : ");
            System.out.println((System.currentTimeMillis() - startTime) / 1000);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
    }

    private static void getFile(String filePath) throws IOException, ClassNotFoundException {

        FileSystem fs = FileSystem.get(URI.create(filePath), HDFSConf.getConf());
        Path path = new Path(filePath);
        if (fs.exists(path) ) {


            FSDataInputStream in;
            FSDataOutputStream out;
            Path outPath;

            FileStatus file = fs.getFileStatus(path);
            // 该压缩方法对应的文件扩展名

            outPath = new Path(filePath.substring(0,filePath.indexOf(".")) + ".new");
            logger.info("out put path is : " + outPath.toString());

            if (fs.createNewFile(outPath)) {

                CompressionCodecFactory factory = new CompressionCodecFactory(HDFSConf.getConf());
                CompressionCodec codec = factory.getCodec(file.getPath());
                in = fs.open(file.getPath());
                InputStream cin = codec.createInputStream(in);
                logger.info("create file  : " + outPath.toString());

                out = fs.append(outPath);

                // 缓冲区设为5MB
                IOUtils.copyBytes(cin, out, 1024 * 1024 * 5, false);

                out.flush();
                cin.close();
                in.close();
                out.close();


                logger.info("Decompress file successful");
            } else {
                logger.error("File exists");
            }


        } else {
            logger.info("There is no file :" + filePath);
        }

    }
}
